<!DOCTYPE html>
<html lang="en">

<head>
	<meta charset="UTF-8">
	<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
	<meta name="keywords" content="seata,seata-golang,seata-go,getty,分布式事务" />
	<meta name="description" content="本文详细讲述 seata-golang 底层 rpc 通信的实现原理" />
	<!-- 网页标签标题 -->
	<title>seata-golang 通信模型详解</title>
  <link rel="shortcut icon" href="/img/seata_logo_small.jpeg"/>
	<link rel="stylesheet" href="/build/blogDetail.css" />
</head>
<body>
	<div id="root"><div class="blog-detail-page" data-reactroot=""><header class="header-container header-container-normal"><div class="header-body"><a href="/zh-cn/index.html"><img class="logo" src="//img.alicdn.com/tfs/TB1gqL1w4D1gK0jSZFyXXciOVXa-1497-401.png"/></a><div class="search search-normal"><span class="icon-search"></span></div><span class="language-switch language-switch-normal">En</span><div class="header-menu"><img class="header-menu-toggle" src="https://img.alicdn.com/tfs/TB14eEmw7P2gK0jSZPxXXacQpXa-38-32.png"/><ul><li class="menu-item menu-item-normal"><a href="/zh-cn/index.html" target="_self">首页</a></li><li class="menu-item menu-item-normal"><a href="/zh-cn/docs/overview/what-is-seata.html" target="_self">文档</a></li><li class="menu-item menu-item-normal"><a href="/zh-cn/docs/developers/developers_dev.html" target="_self">开发者</a></li><li class="menu-item menu-item-normal menu-item-normal-active"><a href="/zh-cn/blog/index.html" target="_self">博客</a></li><li class="menu-item menu-item-normal"><a href="/zh-cn/community/index.html" target="_self">社区</a></li><li class="menu-item menu-item-normal"><a href="/zh-cn/blog/download.html" target="_self">下载</a></li></ul></div></div></header><section class="blog-content markdown-body"><h1>基于 getty 的 seata-golang 通信模型详解</h1>
<p>作者 | 刘晓敏 于雨</p>
<h2>一、简介</h2>
<p>Java 的世界里，大家广泛使用的一个高性能网络通信框架 netty，很多 RPC 框架都是基于 netty 来实现的。在 golang 的世界里，<a href="https://github.com/AlexStocks/getty">getty</a> 也是一个类似 netty 的高性能网络通信库。getty 最初由 dubbogo 项目负责人于雨开发，作为底层通信库在 <a href="https://github.com/apache/dubbo-go">dubbo-go</a> 中使用。随着 dubbo-go 捐献给 apache 基金会，在社区小伙伴的共同努力下，getty 也最终进入到 apache 这个大家庭，并改名 <a href="https://github.com/apache/dubbo-getty">dubbo-getty</a> 。</p>
<p>18 年的时候，我在公司里实践微服务，当时遇到最大的问题就是分布式事务问题。同年，阿里在社区开源他们的分布式事务解决方案，我也很快关注到这个项目，起初还叫 fescar，后来更名 seata。由于我对开源技术很感兴趣，加了很多社区群，当时也很关注 dubbo-go 这个项目，在里面默默潜水。随着对 seata 的了解，逐渐萌生了做一个 go 版本的分布式事务框架的想法。</p>
<p>要做一个 golang 版的分布式事务框架，首要的一个问题就是如何实现 RPC 通信。dubbo-go 就是很好的一个例子摆在眼前，遂开始研究 dubbo-go 的底层 getty。</p>
<h2>二、如何基于 getty 实现 RPC 通信</h2>
<p>getty 框架的整体模型图如下：</p>
<p><img src="https://img.alicdn.com/imgextra/i1/O1CN011TIcL01jY4JaweOfV_!!6000000004559-2-tps-954-853.png" alt="image.png"></p>
<p>下面结合相关代码，详述 seata-golang 的 RPC 通信过程。</p>
<h3>1. 建立连接</h3>
<p>实现 RPC 通信，首先要建立网络连接吧，我们从 <a href="https://github.com/apache/dubbo-getty/blob/master/client.go">client.go</a> 开始看起。</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(c *client)</span> <span class="hljs-title">connect</span><span class="hljs-params">()</span></span> {
	<span class="hljs-keyword">var</span> (
		err error
		ss  Session
	)

	<span class="hljs-keyword">for</span> {
        <span class="hljs-comment">// 建立一个 session 连接</span>
		ss = c.dial()
		<span class="hljs-keyword">if</span> ss == <span class="hljs-literal">nil</span> {
			<span class="hljs-comment">// client has been closed</span>
			<span class="hljs-keyword">break</span>
		}
		err = c.newSession(ss)
		<span class="hljs-keyword">if</span> err == <span class="hljs-literal">nil</span> {
            <span class="hljs-comment">// 收发报文</span>
			ss.(*session).run()
			<span class="hljs-comment">// 此处省略部分代码</span>
      
			<span class="hljs-keyword">break</span>
		}
		<span class="hljs-comment">// don't distinguish between tcp connection and websocket connection. Because</span>
		<span class="hljs-comment">// gorilla/websocket/conn.go:(Conn)Close also invoke net.Conn.Close()</span>
		ss.Conn().Close()
	}
}
</code></pre>
<p><code>connect()</code> 方法通过 <code>dial()</code> 方法得到了一个 session 连接，进入 dial() 方法：</p>
<pre><code>func (c *client) dial() Session {
	switch c.endPointType {
	case TCP_CLIENT:
		return c.dialTCP()
	case UDP_CLIENT:
		return c.dialUDP()
	case WS_CLIENT:
		return c.dialWS()
	case WSS_CLIENT:
		return c.dialWSS()
	}

	return nil
}
</code></pre>
<p>我们关注的是 TCP 连接，所以继续进入 <code>c.dialTCP()</code> 方法：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(c *client)</span> <span class="hljs-title">dialTCP</span><span class="hljs-params">()</span> <span class="hljs-title">Session</span></span> {
	<span class="hljs-keyword">var</span> (
		err  error
		conn net.Conn
	)

	<span class="hljs-keyword">for</span> {
		<span class="hljs-keyword">if</span> c.IsClosed() {
			<span class="hljs-keyword">return</span> <span class="hljs-literal">nil</span>
		}
		<span class="hljs-keyword">if</span> c.sslEnabled {
			<span class="hljs-keyword">if</span> sslConfig, err := c.tlsConfigBuilder.BuildTlsConfig(); err == <span class="hljs-literal">nil</span> &amp;&amp; sslConfig != <span class="hljs-literal">nil</span> {
				d := &amp;net.Dialer{Timeout: connectTimeout}
				<span class="hljs-comment">// 建立加密连接</span>
				conn, err = tls.DialWithDialer(d, <span class="hljs-string">"tcp"</span>, c.addr, sslConfig)
			}
		} <span class="hljs-keyword">else</span> {
            <span class="hljs-comment">// 建立 tcp 连接</span>
			conn, err = net.DialTimeout(<span class="hljs-string">"tcp"</span>, c.addr, connectTimeout)
		}
		<span class="hljs-keyword">if</span> err == <span class="hljs-literal">nil</span> &amp;&amp; gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
			conn.Close()
			err = errSelfConnect
		}
		<span class="hljs-keyword">if</span> err == <span class="hljs-literal">nil</span> {
            <span class="hljs-comment">// 返回一个 TCPSession</span>
			<span class="hljs-keyword">return</span> newTCPSession(conn, c)
		}

		log.Infof(<span class="hljs-string">"net.DialTimeout(addr:%s, timeout:%v) = error:%+v"</span>, c.addr, connectTimeout, perrors.WithStack(err))
		&lt;-wheel.After(connectInterval)
	}
}
</code></pre>
<p>至此，我们知道了 getty 如何建立 TCP 连接，并返回 TCPSession。</p>
<h3>2. 收发报文</h3>
<p>那它是怎么收发报文的呢，我们回到 connection 方法接着往下看，有这样一行 <code>ss.(*session).run()</code>，在这行代码之后代码都是很简单的操作，我们猜测这行代码运行的逻辑里面一定包含收发报文的逻辑，接着进入 <code>run()</code> 方法：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(s *session)</span> <span class="hljs-title">run</span><span class="hljs-params">()</span></span> {
	<span class="hljs-comment">// 省略部分代码</span>
  
	<span class="hljs-keyword">go</span> s.handleLoop()
	<span class="hljs-keyword">go</span> s.handlePackage()
}
</code></pre>
<p>这里起了两个 goroutine，<code>handleLoop</code> 和 <code>handlePackage</code>，看字面意思符合我们的猜想，进入 <code>handleLoop()</code> 方法：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(s *session)</span> <span class="hljs-title">handleLoop</span><span class="hljs-params">()</span></span> {
    <span class="hljs-comment">// 省略部分代码</span>
  
	<span class="hljs-keyword">for</span> {
		<span class="hljs-comment">// A select blocks until one of its cases is ready to run.</span>
		<span class="hljs-comment">// It choose one at random if multiple are ready. Otherwise it choose default branch if none is ready.</span>
		<span class="hljs-keyword">select</span> {
		<span class="hljs-comment">// 省略部分代码</span>
      
		<span class="hljs-keyword">case</span> outPkg, ok = &lt;-s.wQ:
			<span class="hljs-comment">// 省略部分代码</span>

			iovec = iovec[:<span class="hljs-number">0</span>]
			<span class="hljs-keyword">for</span> idx := <span class="hljs-number">0</span>; idx &lt; maxIovecNum; idx++ {
        <span class="hljs-comment">// 通过 s.writer 将 interface{} 类型的 outPkg 编码成二进制的比特</span>
				pkgBytes, err = s.writer.Write(s, outPkg)
				<span class="hljs-comment">// 省略部分代码</span>
        
				iovec = <span class="hljs-built_in">append</span>(iovec, pkgBytes)

                <span class="hljs-comment">//省略部分代码</span>
			}
            <span class="hljs-comment">// 将这些二进制比特发送出去</span>
			err = s.WriteBytesArray(iovec[:]...)
			<span class="hljs-keyword">if</span> err != <span class="hljs-literal">nil</span> {
				log.Errorf(<span class="hljs-string">"%s, [session.handleLoop]s.WriteBytesArray(iovec len:%d) = error:%+v"</span>,
					s.sessionToken(), <span class="hljs-built_in">len</span>(iovec), perrors.WithStack(err))
				s.stop()
				<span class="hljs-comment">// break LOOP</span>
				flag = <span class="hljs-literal">false</span>
			}

		<span class="hljs-keyword">case</span> &lt;-wheel.After(s.period):
			<span class="hljs-keyword">if</span> flag {
				<span class="hljs-keyword">if</span> wsFlag {
					err := wsConn.writePing()
					<span class="hljs-keyword">if</span> err != <span class="hljs-literal">nil</span> {
						log.Warnf(<span class="hljs-string">"wsConn.writePing() = error:%+v"</span>, perrors.WithStack(err))
					}
				}
                <span class="hljs-comment">// 定时执行的逻辑，心跳等</span>
				s.listener.OnCron(s)
			}
		}
	}
}
</code></pre>
<p>通过上面的代码，我们不难发现，<code>handleLoop()</code> 方法处理的是发送报文的逻辑，RPC 需要发送的消息首先由 <code>s.writer</code> 编码成二进制比特，然后通过建立的 TCP 连接发送出去。这个 <code>s.writer</code> 对应的 Writer 接口是 RPC 框架必须要实现的一个接口。</p>
<p>继续看 <code>handlePackage()</code> 方法：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(s *session)</span> <span class="hljs-title">handlePackage</span><span class="hljs-params">()</span></span> {
    <span class="hljs-comment">// 省略部分代码</span>

	<span class="hljs-keyword">if</span> _, ok := s.Connection.(*gettyTCPConn); ok {
		<span class="hljs-keyword">if</span> s.reader == <span class="hljs-literal">nil</span> {
			errStr := fmt.Sprintf(<span class="hljs-string">"session{name:%s, conn:%#v, reader:%#v}"</span>, s.name, s.Connection, s.reader)
			log.Error(errStr)
			<span class="hljs-built_in">panic</span>(errStr)
		}

		err = s.handleTCPPackage()
	} <span class="hljs-keyword">else</span> <span class="hljs-keyword">if</span> _, ok := s.Connection.(*gettyWSConn); ok {
		err = s.handleWSPackage()
	} <span class="hljs-keyword">else</span> <span class="hljs-keyword">if</span> _, ok := s.Connection.(*gettyUDPConn); ok {
		err = s.handleUDPPackage()
	} <span class="hljs-keyword">else</span> {
		<span class="hljs-built_in">panic</span>(fmt.Sprintf(<span class="hljs-string">"unknown type session{%#v}"</span>, s))
	}
}
</code></pre>
<p>进入 <code>handleTCPPackage()</code> 方法：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(s *session)</span> <span class="hljs-title">handleTCPPackage</span><span class="hljs-params">()</span> <span class="hljs-title">error</span></span> {
    <span class="hljs-comment">// 省略部分代码</span>

	conn = s.Connection.(*gettyTCPConn)
	<span class="hljs-keyword">for</span> {
		<span class="hljs-comment">// 省略部分代码</span>

		bufLen = <span class="hljs-number">0</span>
		<span class="hljs-keyword">for</span> {
			<span class="hljs-comment">// for clause for the network timeout condition check</span>
			<span class="hljs-comment">// s.conn.SetReadTimeout(time.Now().Add(s.rTimeout))</span>
            <span class="hljs-comment">// 从 TCP 连接中收到报文</span>
			bufLen, err = conn.recv(buf)
			<span class="hljs-comment">// 省略部分代码</span>
      
			<span class="hljs-keyword">break</span>
		}
		<span class="hljs-comment">// 省略部分代码</span>
    
        <span class="hljs-comment">// 将收到的报文二进制比特写入 pkgBuf</span>
		pktBuf.Write(buf[:bufLen])
		<span class="hljs-keyword">for</span> {
			<span class="hljs-keyword">if</span> pktBuf.Len() &lt;= <span class="hljs-number">0</span> {
				<span class="hljs-keyword">break</span>
			}
            <span class="hljs-comment">// 通过 s.reader 将收到的报文解码成 RPC 消息</span>
			pkg, pkgLen, err = s.reader.Read(s, pktBuf.Bytes())
			<span class="hljs-comment">// 省略部分代码</span>

      s.UpdateActive()
            <span class="hljs-comment">// 将收到的消息放入 TaskQueue 供 RPC 消费端消费</span>
			s.addTask(pkg)
			pktBuf.Next(pkgLen)
			<span class="hljs-comment">// continue to handle case 5</span>
		}
		<span class="hljs-keyword">if</span> exit {
			<span class="hljs-keyword">break</span>
		}
	}

	<span class="hljs-keyword">return</span> perrors.WithStack(err)
}
</code></pre>
<p>从上面的代码逻辑我们分析出，RPC 消费端需要将从 TCP 连接收到的二进制比特报文解码成 RPC 能消费的消息，这个工作由 s.reader 实现，所以，我们要构建 RPC 通信层也需要实现 s.reader 对应的 Reader 接口。</p>
<h3>3. 底层处理网络报文的逻辑如何与业务逻辑解耦</h3>
<p>我们都知道，netty 通过 boss 线程和 worker 线程实现了底层网络逻辑和业务逻辑的解耦。那么，getty 是如何实现的呢？</p>
<p>在 <code>handlePackage()</code> 方法最后，我们看到，收到的消息被放入了 <code>s.addTask(pkg)</code> 这个方法，接着往下分析：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(s *session)</span> <span class="hljs-title">addTask</span><span class="hljs-params">(pkg <span class="hljs-keyword">interface</span>{})</span></span> {
	f := <span class="hljs-function"><span class="hljs-keyword">func</span><span class="hljs-params">()</span></span> {
		s.listener.OnMessage(s, pkg)
		s.incReadPkgNum()
	}
	<span class="hljs-keyword">if</span> taskPool := s.EndPoint().GetTaskPool(); taskPool != <span class="hljs-literal">nil</span> {
		taskPool.AddTaskAlways(f)
		<span class="hljs-keyword">return</span>
	}
	f()
}
</code></pre>
<p><code>pkg</code> 参数传递到了一个匿名方法，这个方法最终放入了 <code>taskPool</code>。这个方法很关键，在我后来写 seata-golang 代码的时候，就遇到了一个坑，这个坑后面分析。</p>
<p>接着我们看一下 <a href="https://github.com/dubbogo/gost/blob/master/sync/task_pool.go">taskPool</a> 的定义：</p>
<pre><code class="language-go"><span class="hljs-comment">// NewTaskPoolSimple build a simple task pool</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-title">NewTaskPoolSimple</span><span class="hljs-params">(size <span class="hljs-keyword">int</span>)</span> <span class="hljs-title">GenericTaskPool</span></span> {
	<span class="hljs-keyword">if</span> size &lt; <span class="hljs-number">1</span> {
		size = runtime.NumCPU() * <span class="hljs-number">100</span>
	}
	<span class="hljs-keyword">return</span> &amp;taskPoolSimple{
		work: <span class="hljs-built_in">make</span>(<span class="hljs-keyword">chan</span> task),
		sem:  <span class="hljs-built_in">make</span>(<span class="hljs-keyword">chan</span> <span class="hljs-keyword">struct</span>{}, size),
		done: <span class="hljs-built_in">make</span>(<span class="hljs-keyword">chan</span> <span class="hljs-keyword">struct</span>{}),
	}
}
</code></pre>
<p>构建了一个缓冲大小为 size （默认为  <code>runtime.NumCPU() * 100</code>） 的 channel <code>sem</code>。再看方法 <code>AddTaskAlways(t task)</code>：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(p *taskPoolSimple)</span> <span class="hljs-title">AddTaskAlways</span><span class="hljs-params">(t task)</span></span> {
	<span class="hljs-keyword">select</span> {
	<span class="hljs-keyword">case</span> &lt;-p.done:
		<span class="hljs-keyword">return</span>
	<span class="hljs-keyword">default</span>:
	}

	<span class="hljs-keyword">select</span> {
	<span class="hljs-keyword">case</span> p.work &lt;- t:
		<span class="hljs-keyword">return</span>
	<span class="hljs-keyword">default</span>:
	}
	<span class="hljs-keyword">select</span> {
	<span class="hljs-keyword">case</span> p.work &lt;- t:
	<span class="hljs-keyword">case</span> p.sem &lt;- <span class="hljs-keyword">struct</span>{}{}:
		p.wg.Add(<span class="hljs-number">1</span>)
		<span class="hljs-keyword">go</span> p.worker(t)
	<span class="hljs-keyword">default</span>:
		goSafely(t)
	}
}
</code></pre>
<p>加入的任务，会先由 len(p.sem) 个 goroutine 去消费，如果没有 goroutine 空闲，则会启动一个临时的 goroutine 去运行 t()。相当于有  len(p.sem) 个 goroutine 组成了 goroutine pool，pool 中的 goroutine 去处理业务逻辑，而不是由处理网络报文的 goroutine 去运行业务逻辑，从而实现了解耦。写 seata-golang 时遇到的一个坑，就是忘记设置 taskPool 造成了处理业务逻辑和处理底层网络报文逻辑的 goroutine 是同一个，我在业务逻辑中阻塞等待一个任务完成时，阻塞了整个 goroutine，使得阻塞期间收不到任何报文。</p>
<h3>4. 具体实现</h3>
<p>下面的代码见 <a href="https://github.com/apache/dubbo-getty/blob/master/getty.go">getty.go</a>：</p>
<pre><code class="language-go"><span class="hljs-comment">// Reader is used to unmarshal a complete pkg from buffer</span>
<span class="hljs-keyword">type</span> Reader <span class="hljs-keyword">interface</span> {
	Read(Session, []<span class="hljs-keyword">byte</span>) (<span class="hljs-keyword">interface</span>{}, <span class="hljs-keyword">int</span>, error)
}

<span class="hljs-comment">// Writer is used to marshal pkg and write to session</span>
<span class="hljs-keyword">type</span> Writer <span class="hljs-keyword">interface</span> {
	<span class="hljs-comment">// if @Session is udpGettySession, the second parameter is UDPContext.</span>
	Write(Session, <span class="hljs-keyword">interface</span>{}) ([]<span class="hljs-keyword">byte</span>, error)
}

<span class="hljs-comment">// ReadWriter interface use for handle application packages</span>
<span class="hljs-keyword">type</span> ReadWriter <span class="hljs-keyword">interface</span> {
	Reader
	Writer
}
</code></pre>
<pre><code class="language-go"><span class="hljs-comment">// EventListener is used to process pkg that received from remote session</span>
<span class="hljs-keyword">type</span> EventListener <span class="hljs-keyword">interface</span> {
	<span class="hljs-comment">// invoked when session opened</span>
	<span class="hljs-comment">// If the return error is not nil, @Session will be closed.</span>
	OnOpen(Session) error

	<span class="hljs-comment">// invoked when session closed.</span>
	OnClose(Session)

	<span class="hljs-comment">// invoked when got error.</span>
	OnError(Session, error)

	<span class="hljs-comment">// invoked periodically, its period can be set by (Session)SetCronPeriod</span>
	OnCron(Session)

	<span class="hljs-comment">// invoked when getty received a package. Pls attention that do not handle long time</span>
	<span class="hljs-comment">// logic processing in this func. You'd better set the package's maximum length.</span>
	<span class="hljs-comment">// If the message's length is greater than it, u should should return err in</span>
	<span class="hljs-comment">// Reader{Read} and getty will close this connection soon.</span>
	<span class="hljs-comment">//</span>
	<span class="hljs-comment">// If ur logic processing in this func will take a long time, u should start a goroutine</span>
	<span class="hljs-comment">// pool(like working thread pool in cpp) to handle the processing asynchronously. Or u</span>
	<span class="hljs-comment">// can do the logic processing in other asynchronous way.</span>
	<span class="hljs-comment">// !!!In short, ur OnMessage callback func should return asap.</span>
	<span class="hljs-comment">//</span>
	<span class="hljs-comment">// If this is a udp event listener, the second parameter type is UDPContext.</span>
	OnMessage(Session, <span class="hljs-keyword">interface</span>{})
}
</code></pre>
<p>通过对整个 getty 代码的分析，我们只要实现  <code>ReadWriter</code> 来对 RPC  消息编解码，再实现 <code>EventListener</code> 来处理 RPC 消息的对应的具体逻辑，将 <code>ReadWriter</code> 实现和 <code>EventLister</code> 实现注入到 RPC 的 Client 和 Server 端，则可实现 RPC 通信。</p>
<h4>4.1 编解码协议实现</h4>
<p>下面是 seata 协议的定义：
<img src="https://cdn.nlark.com/yuque/0/2020/png/737378/1607180799872-5f96afb6-680d-4e69-8c95-b8fd1ac4c3a7.png#align=left&amp;display=inline&amp;height=209&amp;margin=%5Bobject%20Object%5D&amp;name=image-20201205214556457.png&amp;originHeight=209&amp;originWidth=690&amp;size=18407&amp;status=done&amp;style=none&amp;width=690" alt="image-20201205214556457.png"></p>
<p>在 ReadWriter 接口的实现 <a href="https://github.com/opentrx/seata-golang"><code>RpcPackageHandler</code></a> 中，调用 Codec 方法对消息体按照上面的格式编解码：</p>
<pre><code class="language-go"><span class="hljs-comment">// 消息编码为二进制比特</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-title">MessageEncoder</span><span class="hljs-params">(codecType <span class="hljs-keyword">byte</span>, in <span class="hljs-keyword">interface</span>{})</span> []<span class="hljs-title">byte</span></span> {
	<span class="hljs-keyword">switch</span> codecType {
	<span class="hljs-keyword">case</span> SEATA:
		<span class="hljs-keyword">return</span> SeataEncoder(in)
	<span class="hljs-keyword">default</span>:
		log.Errorf(<span class="hljs-string">"not support codecType, %s"</span>, codecType)
		<span class="hljs-keyword">return</span> <span class="hljs-literal">nil</span>
	}
}

<span class="hljs-comment">// 二进制比特解码为消息体</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-title">MessageDecoder</span><span class="hljs-params">(codecType <span class="hljs-keyword">byte</span>, in []<span class="hljs-keyword">byte</span>)</span> <span class="hljs-params">(<span class="hljs-keyword">interface</span>{}, <span class="hljs-keyword">int</span>)</span></span> {
	<span class="hljs-keyword">switch</span> codecType {
	<span class="hljs-keyword">case</span> SEATA:
		<span class="hljs-keyword">return</span> SeataDecoder(in)
	<span class="hljs-keyword">default</span>:
		log.Errorf(<span class="hljs-string">"not support codecType, %s"</span>, codecType)
		<span class="hljs-keyword">return</span> <span class="hljs-literal">nil</span>, <span class="hljs-number">0</span>
	}
}
</code></pre>
<h4>4.2 Client 端实现</h4>
<p>再来看 client 端 <code>EventListener</code> 的实现 <a href="https://github.com/opentrx/seata-golang/blob/dev/pkg/client/rpc_remoting_client.go"><code>RpcRemotingClient</code></a>：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(client *RpcRemoteClient)</span> <span class="hljs-title">OnOpen</span><span class="hljs-params">(session getty.Session)</span> <span class="hljs-title">error</span></span> {
	<span class="hljs-keyword">go</span> <span class="hljs-function"><span class="hljs-keyword">func</span><span class="hljs-params">()</span> 
		<span class="hljs-title">request</span> := <span class="hljs-title">protocal</span>.<span class="hljs-title">RegisterTMRequest</span></span>{AbstractIdentifyRequest: protocal.AbstractIdentifyRequest{
			ApplicationId:           client.conf.ApplicationId,
			TransactionServiceGroup: client.conf.TransactionServiceGroup,
		}}
    <span class="hljs-comment">// 建立连接后向 Transaction Coordinator 发起注册 TransactionManager 的请求</span>
		_, err := client.sendAsyncRequestWithResponse(session, request, RPC_REQUEST_TIMEOUT)
		<span class="hljs-keyword">if</span> err == <span class="hljs-literal">nil</span> {
      <span class="hljs-comment">// 将与 Transaction Coordinator 建立的连接保存在连接池供后续使用</span>
			clientSessionManager.RegisterGettySession(session)
			client.GettySessionOnOpenChannel &lt;- session.RemoteAddr()
		}
	}()

	<span class="hljs-keyword">return</span> <span class="hljs-literal">nil</span>
}

<span class="hljs-comment">// OnError ...</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(client *RpcRemoteClient)</span> <span class="hljs-title">OnError</span><span class="hljs-params">(session getty.Session, err error)</span></span> {
	clientSessionManager.ReleaseGettySession(session)
}

<span class="hljs-comment">// OnClose ...</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(client *RpcRemoteClient)</span> <span class="hljs-title">OnClose</span><span class="hljs-params">(session getty.Session)</span></span> {
	clientSessionManager.ReleaseGettySession(session)
}

<span class="hljs-comment">// OnMessage ...</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(client *RpcRemoteClient)</span> <span class="hljs-title">OnMessage</span><span class="hljs-params">(session getty.Session, pkg <span class="hljs-keyword">interface</span>{})</span></span> {
	log.Info(<span class="hljs-string">"received message:{%v}"</span>, pkg)
	rpcMessage, ok := pkg.(protocal.RpcMessage)
	<span class="hljs-keyword">if</span> ok {
		heartBeat, isHeartBeat := rpcMessage.Body.(protocal.HeartBeatMessage)
		<span class="hljs-keyword">if</span> isHeartBeat &amp;&amp; heartBeat == protocal.HeartBeatMessagePong {
			log.Debugf(<span class="hljs-string">"received PONG from %s"</span>, session.RemoteAddr())
		}
	}

	<span class="hljs-keyword">if</span> rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST ||
		rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST_ONEWAY {
		log.Debugf(<span class="hljs-string">"msgId:%s, body:%v"</span>, rpcMessage.Id, rpcMessage.Body)
      
		<span class="hljs-comment">// 处理事务消息，提交 or 回滚</span>
		client.onMessage(rpcMessage, session.RemoteAddr())
	} <span class="hljs-keyword">else</span> {
		resp, loaded := client.futures.Load(rpcMessage.Id)
		<span class="hljs-keyword">if</span> loaded {
			response := resp.(*getty2.MessageFuture)
			response.Response = rpcMessage.Body
			response.Done &lt;- <span class="hljs-literal">true</span>
			client.futures.Delete(rpcMessage.Id)
		}
	}
}

<span class="hljs-comment">// OnCron ...</span>
<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(client *RpcRemoteClient)</span> <span class="hljs-title">OnCron</span><span class="hljs-params">(session getty.Session)</span></span> {
  <span class="hljs-comment">// 发送心跳</span>
	client.defaultSendRequest(session, protocal.HeartBeatMessagePing)
}
</code></pre>
<p><code>clientSessionManager.RegisterGettySession(session)</code> 的逻辑 4.4 小节分析。</p>
<h4>4.3 Server 端 Transaction Coordinator 实现</h4>
<p>代码见 <a href="https://github.com/opentrx/seata-golang/blob/dev/tc/server/default_coordinator_event_listener.go"><code>DefaultCoordinator</code></a>：</p>
<pre><code class="language-go"><span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(coordinator *DefaultCoordinator)</span> <span class="hljs-title">OnOpen</span><span class="hljs-params">(session getty.Session)</span> <span class="hljs-title">error</span></span> {
	log.Infof(<span class="hljs-string">"got getty_session:%s"</span>, session.Stat())
	<span class="hljs-keyword">return</span> <span class="hljs-literal">nil</span>
}

<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(coordinator *DefaultCoordinator)</span> <span class="hljs-title">OnError</span><span class="hljs-params">(session getty.Session, err error)</span></span> {
	<span class="hljs-comment">// 释放 TCP 连接</span>
  SessionManager.ReleaseGettySession(session)
	session.Close()
	log.Errorf(<span class="hljs-string">"getty_session{%s} got error{%v}, will be closed."</span>, session.Stat(), err)
}

<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(coordinator *DefaultCoordinator)</span> <span class="hljs-title">OnClose</span><span class="hljs-params">(session getty.Session)</span></span> {
	log.Info(<span class="hljs-string">"getty_session{%s} is closing......"</span>, session.Stat())
}

<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(coordinator *DefaultCoordinator)</span> <span class="hljs-title">OnMessage</span><span class="hljs-params">(session getty.Session, pkg <span class="hljs-keyword">interface</span>{})</span></span> {
	log.Debugf(<span class="hljs-string">"received message:{%v}"</span>, pkg)
	rpcMessage, ok := pkg.(protocal.RpcMessage)
	<span class="hljs-keyword">if</span> ok {
		_, isRegTM := rpcMessage.Body.(protocal.RegisterTMRequest)
		<span class="hljs-keyword">if</span> isRegTM {
      <span class="hljs-comment">// 将 TransactionManager 信息和 TCP 连接建立映射关系</span>
			coordinator.OnRegTmMessage(rpcMessage, session)
			<span class="hljs-keyword">return</span>
		}

		heartBeat, isHeartBeat := rpcMessage.Body.(protocal.HeartBeatMessage)
		<span class="hljs-keyword">if</span> isHeartBeat &amp;&amp; heartBeat == protocal.HeartBeatMessagePing {
			coordinator.OnCheckMessage(rpcMessage, session)
			<span class="hljs-keyword">return</span>
		}

		<span class="hljs-keyword">if</span> rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST ||
			rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST_ONEWAY {
			log.Debugf(<span class="hljs-string">"msgId:%s, body:%v"</span>, rpcMessage.Id, rpcMessage.Body)
			_, isRegRM := rpcMessage.Body.(protocal.RegisterRMRequest)
			<span class="hljs-keyword">if</span> isRegRM {
        <span class="hljs-comment">// 将 ResourceManager 信息和 TCP 连接建立映射关系</span>
				coordinator.OnRegRmMessage(rpcMessage, session)
			} <span class="hljs-keyword">else</span> {
				<span class="hljs-keyword">if</span> SessionManager.IsRegistered(session) {
					<span class="hljs-keyword">defer</span> <span class="hljs-function"><span class="hljs-keyword">func</span><span class="hljs-params">()</span></span> {
						<span class="hljs-keyword">if</span> err := <span class="hljs-built_in">recover</span>(); err != <span class="hljs-literal">nil</span> {
							log.Errorf(<span class="hljs-string">"Catch Exception while do RPC, request: %v,err: %w"</span>, rpcMessage, err)
						}
					}()
          <span class="hljs-comment">// 处理事务消息，全局事务注册、分支事务注册、分支事务提交、全局事务回滚等</span>
					coordinator.OnTrxMessage(rpcMessage, session)
				} <span class="hljs-keyword">else</span> {
					session.Close()
					log.Infof(<span class="hljs-string">"close a unhandled connection! [%v]"</span>, session)
				}
			}
		} <span class="hljs-keyword">else</span> {
			resp, loaded := coordinator.futures.Load(rpcMessage.Id)
			<span class="hljs-keyword">if</span> loaded {
				response := resp.(*getty2.MessageFuture)
				response.Response = rpcMessage.Body
				response.Done &lt;- <span class="hljs-literal">true</span>
				coordinator.futures.Delete(rpcMessage.Id)
			}
		}
	}
}

<span class="hljs-function"><span class="hljs-keyword">func</span> <span class="hljs-params">(coordinator *DefaultCoordinator)</span> <span class="hljs-title">OnCron</span><span class="hljs-params">(session getty.Session)</span></span> {

}
</code></pre>
<p><code>coordinator.OnRegTmMessage(rpcMessage, session)</code> 注册 Transaction Manager，<code>coordinator.OnRegRmMessage(rpcMessage, session)</code> 注册 Resource Manager。具体逻辑分析见 4.4 小节。
消息进入 <code>coordinator.OnTrxMessage(rpcMessage, session)</code> 方法，将按照消息的类型码路由到具体的逻辑当中：</p>
<pre><code class="language-go">	<span class="hljs-keyword">switch</span> msg.GetTypeCode() {
	<span class="hljs-keyword">case</span> protocal.TypeGlobalBegin:
		req := msg.(protocal.GlobalBeginRequest)
		resp := coordinator.doGlobalBegin(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">case</span> protocal.TypeGlobalStatus:
		req := msg.(protocal.GlobalStatusRequest)
		resp := coordinator.doGlobalStatus(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">case</span> protocal.TypeGlobalReport:
		req := msg.(protocal.GlobalReportRequest)
		resp := coordinator.doGlobalReport(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">case</span> protocal.TypeGlobalCommit:
		req := msg.(protocal.GlobalCommitRequest)
		resp := coordinator.doGlobalCommit(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">case</span> protocal.TypeGlobalRollback:
		req := msg.(protocal.GlobalRollbackRequest)
		resp := coordinator.doGlobalRollback(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">case</span> protocal.TypeBranchRegister:
		req := msg.(protocal.BranchRegisterRequest)
		resp := coordinator.doBranchRegister(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">case</span> protocal.TypeBranchStatusReport:
		req := msg.(protocal.BranchReportRequest)
		resp := coordinator.doBranchReport(req, ctx)
		<span class="hljs-keyword">return</span> resp
	<span class="hljs-keyword">default</span>:
		<span class="hljs-keyword">return</span> <span class="hljs-literal">nil</span>
	}
</code></pre>
<h4>4.4 session manager 分析</h4>
<p>Client 端同 Transaction Coordinator 建立连接起连接后，通过 <code>clientSessionManager.RegisterGettySession(session)</code> 将连接保存在 <code>serverSessions = sync.Map{}</code> 这个 map 中。map 的 key 为从 session 中获取的 RemoteAddress 即 Transaction Coordinator 的地址，value 为 session。这样，Client 端就可以通过 map 中的一个 session 来向 Transaction Coordinator 注册 Transaction Manager 和 Resource Manager 了。具体代码见 <a href="https://github.com/opentrx/seata-golang/blob/dev/pkg/client/getty_client_session_manager.go"><code>getty_client_session_manager.go</code>。</a>
Transaction Manager 和 Resource Manager 注册到 Transaction Coordinator 后，一个连接既有可能用来发送 TM 消息也有可能用来发送 RM 消息。我们通过 RpcContext 来标识一个连接信息：</p>
<pre><code class="language-go"><span class="hljs-keyword">type</span> RpcContext <span class="hljs-keyword">struct</span> {
	Version                 <span class="hljs-keyword">string</span>
	TransactionServiceGroup <span class="hljs-keyword">string</span>
	ClientRole              meta.TransactionRole
	ApplicationId           <span class="hljs-keyword">string</span>
	ClientId                <span class="hljs-keyword">string</span>
	ResourceSets            *model.Set
	Session                 getty.Session
}
</code></pre>
<p>当收到事务消息时，我们需要构造这样一个 RpcContext 供后续事务处理逻辑使用。所以，我们会构造下列 map 来缓存映射关系：</p>
<pre><code class="language-go"><span class="hljs-keyword">var</span> (
	<span class="hljs-comment">// session -&gt; transactionRole</span>
	<span class="hljs-comment">// TM will register before RM, if a session is not the TM registered,</span>
	<span class="hljs-comment">// it will be the RM registered</span>
	session_transactionroles = sync.Map{}

	<span class="hljs-comment">// session -&gt; applicationId</span>
	identified_sessions = sync.Map{}

	<span class="hljs-comment">// applicationId -&gt; ip -&gt; port -&gt; session</span>
	client_sessions = sync.Map{}

	<span class="hljs-comment">// applicationId -&gt; resourceIds</span>
	client_resources = sync.Map{}
)
</code></pre>
<p>这样，Transaction Manager 和 Resource Manager 分别通过 <code>coordinator.OnRegTmMessage(rpcMessage, session)</code> 和 <code>coordinator.OnRegRmMessage(rpcMessage, session)</code> 注册到 Transaction Coordinator 时，会在上述 client_sessions map 中缓存 applicationId、ip、port 与 session 的关系，在 client_resources map 中缓存 applicationId 与 resourceIds（一个应用可能存在多个 Resource Manager） 的关系。在需要时，我们就可以通过上述映射关系构造一个 RpcContext。这部分的实现和 java 版 seata 有很大的不同，感兴趣的可以深入了解一下。具体代码见 <a href="https://github.com/opentrx/seata-golang/blob/dev/tc/server/getty_session_manager.go"><code>getty_session_manager.go</code>。</a>
至此，我们就分析完了 <a href="https://github.com/opentrx/seata-golang">seata-golang</a> 整个 RPC 通信模型的机制。</p>
<h2>三、seata-golang 的未来</h2>
<p><a href="https://github.com/opentrx/seata-golang">seata-golang</a>  从今年 4 月份开始开发，到 8 月份基本实现和 java 版 <a href="https://github.com/seata/seata">seata 1.2</a> 协议的互通，对 mysql 数据库实现了 AT 模式（自动协调分布式事务的提交回滚），实现了 TCC 模式，TC 端使用 mysql 存储数据，使 TC 变成一个无状态应用支持高可用部署。下图展示了 AT 模式的原理：<img src="https://img.alicdn.com/imgextra/i3/O1CN01alqsQS1G2oQecFYIs_!!6000000000565-2-tps-1025-573.png" alt="image20201205-232516.png"></p>
<p>后续，还有许多工作可以做，比如：对注册中心的支持、对配置中心的支持、和 java 版 seata 1.4 的协议互通、其他数据库的支持、raft transaction coordinator 的实现等，希望对分布式事务问题感兴趣的开发者可以加入进来一起来打造一个完善的 golang 的分布式事务框架。</p>
<p>如果你有任何疑问，欢迎钉钉扫码加入交流群【钉钉群号 33069364】：
<img src="https://img.alicdn.com/imgextra/i2/O1CN01IjOVG425erjuzqcOi_!!6000000007552-2-tps-600-621.png" width="200px" /></p>
<h3><strong>作者简介</strong></h3>
<p>刘晓敏 (GitHubID dk-lockdown)，目前就职于 h3c 成都分公司，擅长使用 Java/Go 语言，在云原生和微服务相关技术方向均有涉猎，目前专攻分布式事务。
于雨(github @AlexStocks)，dubbo-go 项目和社区负责人，一个有十多年服务端基础架构研发一线工作经验的程序员，陆续参与改进过 Muduo/Pika/Dubbo/Sentinel-go 等知名项目，目前在蚂蚁金服可信原生部从事容器编排和 service mesh 工作。</p>
<h4>参考资料</h4>
<p>seata 官方：<a href="https://seata.io">https://seata.io</a></p>
<p>java 版 seata：<a href="https://github.com/seata/seata">https://github.com/seata/seata</a></p>
<p>seata-golang 项目地址：<a href="https://github.com/transaction-wg/seata-golang">https://github.com/opentrx/seata-golang</a></p>
<p>seata-golang go 夜读 b站分享：<a href="https://www.bilibili.com/video/BV1oz411e72T">https://www.bilibili.com/video/BV1oz411e72T</a></p>
</section><footer class="footer-container"><div class="footer-body"><img src="//img.alicdn.com/tfs/TB1dGrSwVT7gK0jSZFpXXaTkpXa-4802-1285.png"/><p class="docsite-power">website powered by docsite</p><div class="cols-container"><div class="col col-12"><h3>愿景</h3><p>Seata 是一款阿里巴巴开源的分布式事务解决方案，致力于在微服务架构下提供高性能和简单易用的分布式事务服务。</p></div><div class="col col-6"><dl><dt>文档</dt><dd><a href="/zh-cn/docs/overview/what-is-seata.html" target="_self">Seata 是什么？</a></dd><dd><a href="/zh-cn/docs/user/quickstart.html" target="_self">快速开始</a></dd><dd><a href="https://github.com/seata/seata.github.io/issues/new" target="_self">报告文档问题</a></dd><dd><a href="https://github.com/seata/seata.github.io" target="_self">在Github上编辑此文档</a></dd></dl></div><div class="col col-6"><dl><dt>资源</dt><dd><a href="/zh-cn/blog/index.html" target="_self">博客</a></dd><dd><a href="/zh-cn/community/index.html" target="_self">社区</a></dd></dl></div></div><div class="copyright"><span>Copyright © 2019 Seata</span></div></div></footer></div></div>
	<script src="https://f.alicdn.com/react/15.4.1/react-with-addons.min.js"></script>
	<script src="https://f.alicdn.com/react/15.4.1/react-dom.min.js"></script>
	<script>
		window.rootPath = '';
  </script>
	<script src="/build/blogDetail.js"></script>
	<script>
    var _hmt = _hmt || [];
    (function() {
      var hm = document.createElement("script");
      hm.src = "https://hm.baidu.com/hm.js?104e73ef0c18b416b27abb23757ed8ee";
      var s = document.getElementsByTagName("script")[0];
      s.parentNode.insertBefore(hm, s);
    })();
    </script>
</body>
</html>
